package com.simple.datax.sender.impl;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.SimpleFailMessage;
import com.simple.datax.SimpleMessage;
import com.simple.datax.SimpleObject;
import com.simple.datax.api.ConnectorException;
import com.simple.datax.api.Sender;
import com.simple.datax.conn.BaseConnector;
import com.simple.datax.conn.ConnectorPool;
import com.simple.datax.conn.impl.ConnectorPoolManager;
import com.simple.datax.conn.impl.SendThreadPoolExecutor;
import com.simple.datax.core.model.connpara.CommonConnPara;
import com.simple.datax.service.impl.SendStatusHandler;
import com.simple.datax.utils.MessageUtils;
import com.simple.datax.utils.SimpleConstants;

public class SenderImpl implements Sender {
	private static final Logger logger = LoggerFactory
			.getLogger(SenderImpl.class);

	private SendStatusHandler handler = null;
	private ConnectorPool connectorPool;
	private SendThreadPoolExecutor executor;
	
	public void init() {
		connectorPool = ConnectorPoolManager.getInstance().getConnectorPool(ConnectorPoolManager.CONN_SEND, handler, executor);
	}
	
	/**
	 * All processor can call this function to send message out.
	 * This function returns without block and no exception throw out.
	 * If any issue happens, save status to DB and logger into error log. 
	 */
	@Override
	public void send(SimpleObject src) {
		if (src instanceof SimpleMessage) {
			String connPara = ((SimpleMessage) src).getSentCommunParam();
			String connType = ((SimpleMessage) src).getSentCommunType();
			String encoding = ((SimpleMessage) src).getSentEncoding();
			
			BaseConnector connector = null;
			try {
				CommonConnPara para = new CommonConnPara();
				para.setEncoding(encoding);
				para.setTransportType(connType);
				para.setTransportPara(connPara);
				connector = connectorPool.borrowObject(para);
				
				//for FTP, MSMQ, ActiveMQ test
//				connector = connectorPool.borrowObject(connType, connPara+"_sender");
				// for IBMMQ test ONLY
//				connPara = "10.2.3.52$1500$mqadmin$Art515940$1381$RemoteSend$JAVA.CHANNEL$ReceiveQueue";
//				connector = connectorPool.borrowObject(connType, connPara);
				
				if (connector != null) {
					connector.send(src);
				} else {
					handleError((SimpleMessage)src, new ConnectorException("No available connector for sending message"));
				}
			} catch (Exception e) {
				handleError((SimpleMessage)src, e);
			}  finally {
				if (connector != null) {
					try {
						connectorPool.returnObject(connector);
					} catch (Exception e) {
						logger.error("Connector release error.", e);
					}
				}
			}
		} else {
			handleError((SimpleMessage)src, new InvalidSimpleObjectException(src.getClass().getName()));
		}
	}
	
	/**
	 * handle error in sending message.
	 * 1) callFailService
	 * 2) callResponseService
	 * @param src
	 * @param ex
	 */
	private void handleError(SimpleMessage src, Exception ex) {
		logger.error("Fail to send out message.", ex);
		String errorMessage = ex.getMessage();
		try {
			// 失败处理
			SimpleFailMessage sfm = MessageUtils.castFailMessage(
					errorMessage, SimpleConstants.ERROR_TYPE_SEND,
					SimpleConstants.ERROR_LEVEL_NORMAL,
					src);

			handler.callFailService(sfm);// 注意这里需要处理SimpleFailMessage
		} catch (Exception e) {
			logger.error("Sender FailMessage Error:", e);
		}

		try {
			// 处理错误response报文
			int responseCode = SimpleConstants.RESPONSE_CODE_SENDER;
			String responseMsg = SimpleConstants.RESPONSE_MSG_SENDER;
			handler.callResponseService(src, responseCode,
					responseMsg, src.isNeedResponse());
		} catch (Exception e) {
			logger.error("Sender FailMessage Response Error:", e);
		}
	}

	public SendStatusHandler getHandler() {
		return handler;
	}

	public void setHandler(SendStatusHandler handler) {
		this.handler = handler;
	}
	public SendThreadPoolExecutor getExecutor() {
		return executor;
	}
	public void setExecutor(SendThreadPoolExecutor executor) {
		this.executor = executor;
	}
}
